package com.zx.sms.connect.manager;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.zx.sms.BaseMessage;
import com.zx.sms.codec.cmpp.wap.LongMessageMarkerReadHandler;
import com.zx.sms.common.GlobalConstance;
import com.zx.sms.common.storedMap.BDBStoredMapFactoryImpl;
import com.zx.sms.common.storedMap.VersionObject;
import com.zx.sms.connect.manager.cmpp.CMPPServerEndpointEntity;
import com.zx.sms.handler.MessageLogHandler;
import com.zx.sms.handler.api.AbstractBusinessHandler;
import com.zx.sms.handler.api.BusinessHandlerInterface;
import com.zx.sms.session.AbstractSessionStateManager;
import com.zx.sms.session.cmpp.SessionState;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.traffic.WindowSizeChannelTrafficShapingHandler;
import io.netty.util.concurrent.Promise;

/**
 * @author Lihuanghe(18852780@qq.com)
 */
public abstract class AbstractEndpointConnector implements EndpointConnector<EndpointEntity> {
	private static final Logger logger = LoggerFactory.getLogger(AbstractEndpointConnector.class);

	/**
	 * 端口
	 */
	private EndpointEntity endpoint;
	
	private static final String BDBPREFIX = "Session_" ;

	private CircularList channels = new CircularList();

	public AbstractEndpointConnector(EndpointEntity endpoint) {
		this.endpoint = endpoint;
	}

	@Override
	public EndpointEntity getEndpointEntity() {

		return endpoint;
	}

	@Override
	public void close(Channel channel) throws Exception {
		if (channel.isOpen())
			channel.close();
		// 将channel移除
		removeChannel(channel);
	}

	@Override
	public void close() throws Exception {
		Channel[] chs = channels.getall();
		if (chs == null || chs.length == 0)
			return;
		for (Channel ch : chs) {
			close(ch);
		}
		if (endpoint.isReSendFailMsg()) {
			BDBStoredMapFactoryImpl.INS.close(endpoint.getId(),BDBPREFIX + endpoint.getId());
		}
	}

	@Override
	public Channel fetch() {
		Channel ch = channels.fetch();

		if (ch != null && ch.isActive()) {
			if (ch.attr(GlobalConstance.attributeKey).get() == SessionState.Connect) {
				return ch;
			}
		}
		return null;
	}

	@Override
	public int getConnectionNum() {

		return getChannels().size();
	}

	private CircularList getChannels() {
		return channels;
	}

	protected abstract AbstractSessionStateManager createSessionManager(EndpointEntity entity, ConcurrentMap storeMap, boolean presend);

	protected abstract void doBindHandler(ChannelPipeline pipe, EndpointEntity entity);

	public synchronized boolean addChannel(Channel ch) {
		int nowConnCnt = getConnectionNum();
		EndpointEntity endpoint = getEndpointEntity();
		if (endpoint.getMaxChannels() == 0 || endpoint.getMaxChannels() > nowConnCnt) {

			ConcurrentMap<Serializable, VersionObject> storedMap = null;
			if (endpoint.isReSendFailMsg()) {
				// 如果上次发送失败的消息要重发一次，则要创建持久化Map用于存储发送的message
				storedMap = BDBStoredMapFactoryImpl.INS.buildMap(endpoint.getId(), BDBPREFIX + endpoint.getId());
			} else {
				storedMap = new ConcurrentHashMap();
			}

			logger.info("Channel added To Endpoint {} .totalCnt:{} ,remoteAddress: {}", endpoint, nowConnCnt + 1, ch.remoteAddress());
			
			// 如果是第一个连接。要把上次发送失败的消息取出，再次发送一次
			AbstractSessionStateManager sessionManager = createSessionManager(endpoint, storedMap, nowConnCnt == 0 && endpoint.isReSendFailMsg());
			
			ch.pipeline().addAfter(GlobalConstance.codecName, GlobalConstance.sessionStateManager, sessionManager);
			// 增加流量整形 ，每个连接每秒发送，接收消息数不超过配置的值
			ch.pipeline().addAfter(GlobalConstance.codecName, "ChannelTrafficAfter",
					new WindowSizeChannelTrafficShapingHandler(endpoint, 100));
			
			ch.pipeline().addAfter(GlobalConstance.codecName, "msgLog", new MessageLogHandler(endpoint));
			
			//添加长短信标识Handler : LongMessageMarkerHandler
			//用于给长短信类型的msg打上唯一标识
			LongMessageMarkerReadHandler h_readMarker = new LongMessageMarkerReadHandler(endpoint);
			ch.pipeline().addAfter(GlobalConstance.codecName, h_readMarker.name(),h_readMarker );

			//添加各个协议的业务Handler
			bindHandler(ch.pipeline(), endpoint);
			
			/* 所有的handler都已加入pipeline后再标识连接已建立，
			 * 如过早加入connector，遇到有消息发送时，可能业务handler还未加入，
			 * 引起消息未经handler处理就发了出去。
			 */
			ch.attr(GlobalConstance.attributeKey).set(SessionState.Connect);
			ch.attr(GlobalConstance.entityPointKey).set(endpoint);
			
			ch.attr(GlobalConstance.sessionKey).set(sessionManager);
			ch.attr(GlobalConstance.channelActiveTime).set(System.currentTimeMillis());
			getChannels().add(ch);
			return true;
		} else {
			logger.warn("allowed max channel count: {} ,deny to login.{}", endpoint.getMaxChannels(), endpoint);
			return false;
		}

	}

	public void removeChannel(Channel ch) {

		if (getChannels().remove(ch)) {
			ch.attr(GlobalConstance.attributeKey).set(SessionState.DisConnect);
			ch.attr(GlobalConstance.sessionKey).set(null);
			//以下属性后边handler可能还要使用，不清除
//			ch.attr(GlobalConstance.entityPointKey).set(null);
//			ch.attr(GlobalConstance.channelActiveTime).set(null);
		}
		synchronized(this) {
			if( endpoint.isReSendFailMsg() && getChannels().size() == 0 ) {
				BDBStoredMapFactoryImpl.INS.close(endpoint.getId(),BDBPREFIX + endpoint.getId());
			}
		}
	}

	/**
	 * 连接建立成功后要加载的channelHandler
	 */
	protected void bindHandler(ChannelPipeline pipe, EndpointEntity entity) {

		if (entity instanceof CMPPServerEndpointEntity) {
			return;
		}
		pipe.addBefore(GlobalConstance.IdleCheckerHandlerName,"socketLog", new LoggingHandler(String.format(GlobalConstance.loggerNamePrefix, entity.getId()), LogLevel.TRACE));
		// 调用子类的bind方法
		doBindHandler(pipe, entity);

		List<BusinessHandlerInterface> handlers = entity.getBusinessHandlerSet();
		if (handlers != null && handlers.size() > 0) {
			for (BusinessHandlerInterface handler : handlers) {
				if (handler instanceof AbstractBusinessHandler) {
					AbstractBusinessHandler buziHandler = (AbstractBusinessHandler) handler;
					buziHandler.setEndpointEntity(entity);
					if (buziHandler.isSharable()) {
						pipe.addLast(buziHandler.name(), buziHandler);
					} else {
						AbstractBusinessHandler cloned = null;
						try {
							cloned = buziHandler.clone();

						} catch (CloneNotSupportedException e) {
							logger.error("handlers is not shareable and not implements Cloneable", e);
						}
						if (cloned != null) {
							cloned.setEndpointEntity(entity);
							pipe.addLast(buziHandler.name(), cloned);
							logger.info("handlers is not shareable . clone it success. {}", cloned);
						}
					}

				} else {
					handler.setEndpointEntity(entity);
					pipe.addLast(handler.name(), handler);
					logger.info("add share handlers . {}", handler);
				}
			}
		}
		
		ChannelInitializer bizInitializer = entity.getBusinessChannelInitializer();
		if(bizInitializer != null)
			pipe.addLast(bizInitializer);
		// 黑洞处理，丢弃所有消息
		pipe.addLast("BlackHole", GlobalConstance.blackhole);

	}

	public Channel[] getallChannel() {
		return channels.getall();
	}

	/**
	 * 循环列表，用于实现轮循算法
	 */
	private class CircularList {
		private List<Channel> collection = Collections.synchronizedList(new ArrayList<Channel>());

		public Channel[] getall() {
			return collection.toArray(new Channel[collection.size()]);
		}

		public int size() {
			return collection.size();
		}

		public Channel fetch() {
			int size = collection.size();
			if (size == 0)
				return null;
			
			if (size == 1)
				return collection.get(0);

			int idx = indexSeq.getAndIncrement() & 0xfffff; //避免为负数

			try {
				Channel ret = collection.get(idx%size);
				return ret;
			} catch (IndexOutOfBoundsException ex) {
				// 多线程情况可能抛异常
				// 1：当线连接数为0了
				// 2：当前连接数小于index
				indexSeq.set(0);
				return collection.isEmpty() ? null : collection.get(0);
			} finally {
			}
		}

		public boolean add(Channel ele) {

			boolean r = false;
			try {
				r = collection.add(ele);
			} finally {
			}
			return r;
		}

		public boolean remove(Channel ele) {

			boolean r = false;
			try {
				r = collection.remove(ele);
			} finally {
			}
			return r;
		}

		private AtomicInteger indexSeq = new AtomicInteger();
	}

	public ChannelFuture asynwriteUncheck(Object msg) {
		Channel ch = fetch();
		if (ch == null)
			return null;
		ChannelFuture future = ch.writeAndFlush(msg);
		return future;
	}
	
	public ChannelFuture asynwrite(Object msg) {
		Channel ch = fetchOneWritable();
		if (ch == null)
			return null;
		ChannelFuture future = ch.writeAndFlush(msg);
		return future;
	}
	
	public <T extends BaseMessage> Promise<T> synwriteUncheck(T message) {
		return synwrite(message,false);
	}
	
	public <T extends BaseMessage> List<Promise<T>> synwriteUncheck(List<T> msgs) {
		return synwrite(msgs,false);
	}
	
	public <T extends BaseMessage> List<Promise<T>> synwrite(List<T> msgs) {
		return synwrite(msgs,true);
	}
	public <T extends BaseMessage> Promise<T> synwrite(T message) {
		return synwrite(message,true);
	}
	

	private <T extends BaseMessage> List<Promise<T>> synwrite(List<T> msgs,boolean checkWritable) {
		Channel ch = checkWritable ? fetchOneWritable() : fetch();
		if (ch == null)
			return null;
		AbstractSessionStateManager session =  ch.attr(GlobalConstance.sessionKey).get();
		if (session == null)
			return null;
		List<Promise<T>> arrPromise = new ArrayList<Promise<T>>();
		for (BaseMessage msg : msgs) {
			arrPromise.add(session.writeMessagesync(msg));
		}

		return arrPromise;
	}

	private <T extends BaseMessage> Promise<T> synwrite(T message,boolean checkWritable) {
		Channel ch = checkWritable ? fetchOneWritable() : fetch();
		if (ch == null)
			return null;
		AbstractSessionStateManager session = ch.attr(GlobalConstance.sessionKey).get();
		if (session == null)
			return null;
		return session.writeMessagesync(message);
	}

	private Channel fetchOneWritable() {
		Channel ch = fetch();
		// 端口上还没有可用连接
		if (ch == null)
			return null;

		if (ch.isActive() && ch.isWritable()) {
			return ch;
		}
		return null;
	}

}
